/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
/**
 * Software License, Version 1.0
 *
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 *
 *
 *Redistribution and use in source and binary forms, with or without
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For
 * further information contact the Community Grids Lab at
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and
 * wherever such third-party acknowledgments normally appear.
 *
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering,
 * shall not be used to endorse or promote products derived from this software
 * without prior written permission from Indiana University.  For written
 * permission, please contact the Advanced Research and Technology Institute
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering,
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 *
 *
 * Indiana University provides no reassurances that the source code provided
 * does not infringe the patent or any other intellectual property rights of
 * any other entity.  Indiana University disclaims any liability to any
 * recipient for claims brought by any other entity based on infringement of
 * intellectual property rights or otherwise.
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS.
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS",
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.transport.sslHttpBase;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.text.NumberFormat;
import java.util.Enumeration;
import java.util.Properties;
import java.util.Random;
import java.util.Vector;

import cgl.narada.performance.PerformanceMeasurement;
import cgl.narada.performance.PerformanceMeasurementListener;
import cgl.narada.transport.Link;
import cgl.narada.transport.MonitoredLink;
import cgl.narada.transport.TransportException;
import cgl.narada.transport.util.BytesUtil;
import cgl.narada.util.DataQueue;
import cgl.narada.util.logging.Logger;
import cgl.narada.util.logging.LoggerFactory;



/**
 * The Link implementation which serves as a base for SSL communications.
 * @author John Yin and Hongbin Liu
 * $Date$ $Revision$
 */

public class LinkImpl extends MonitoredLink implements Link, PerformanceMeasurementListener {

    final static Logger log =
	LoggerFactory.getLogger(LinkImpl.class.getName());

    // Used to generated random seed for link id,
    private static Random    s_rand = new Random(System.currentTimeMillis());
    private static NumberFormat s_numberFormat = NumberFormat.getInstance();
    /**
     * The performance parameter names to return to the appliaction.
     */
//    public static final String[] PERF_MEASUREMENTS = {
//        "max data rate",
//        "avg. latency",
//        "min. latency",
//        "max. latency",
//        "jitter",
//        "message rate",
//        "throughput"
//    };

    /**
     * The performance parameter names to return to the appliaction.
     */
//    private static final String[] PERF_MEASUREMENT_NAMES = {
//        "kb/sec",
//        "latency",
//        "latency",
//        "latency",
//        "jitter",
//        "msg/sec",
//        "throughput"
//    };

//    private static final byte[] PERF_STAT_TYPE = {
//        StatCollection.TIME_AVG_STAT,
//        StatCollection.AVG_STAT,
//        StatCollection.MIN_STAT,
//        StatCollection.MAX_STAT,
//        StatCollection.MAX_STAT,
//        StatCollection.TIME_AVG_STAT,
//        StatCollection.AVG_STAT,
//    };

//    private LinkPerformanceDataImpl[] m_perfData = {
//        new LinkPerformanceDataImpl("data rate","Application data send rate","KB/sec"),
//        new LinkPerformanceDataImpl("avg. latency","Average round trip latency","ms"),
//        new LinkPerformanceDataImpl("min. latency","Minimum round trip latency","ms"),
//        new LinkPerformanceDataImpl("max. latency","Maximum round trip latency","ms"),
//        new LinkPerformanceDataImpl("jitter","Jitter","ms"),
//        new LinkPerformanceDataImpl("message rate","Application message send rate","msg/sec"),
//        new LinkPerformanceDataImpl("throughput","Est. maximum data send rate","KB/sec"),
//    };

    /**
     * -------------------------------------------------------------------
     */
    private Transport m_transport = null;
    protected String    m_linkID = null;
    private Object    m_brokerID  = null;
    private DataQueue m_queue     = new DataQueue();
    private String    m_linkType = null;
    private boolean   m_closed      = false;
    private LinkManager m_parent = null;
    // Used for having a keep alive link
    private long      m_keepAliveStartTime = 0;
    private long      m_keepAliveEndTime   = 0;
    private boolean   m_performanceEnabled = false;

    private Properties m_creationProperties = null;

    /**
     * -------------------------------------------------------------------
     */
    protected LinkImpl(Transport transport,String linkType, LinkManager parent) {
        m_transport = transport;
        m_linkID = transport.toString()+":"+Math.abs(s_rand.nextInt());
        m_linkType = linkType;
        m_parent      = parent;
        s_numberFormat.setMaximumFractionDigits(2);
    }

    /**
     * -------------------------------------------------------------------
     */
    public void sendData(byte[] data) throws TransportException {
        if (m_closed) {
            throw new TransportException("Link already closed");
        }

        LinkMessage message = new LinkMessage(m_linkID,data);
        m_parent.getPerformanceMeasurement().updatePSMetric(getName(),"msg/sec",1);
        m_parent.getPerformanceMeasurement().updatePSMetric(getName(),"kb/sec",(double)data.length/1000);
        try {
            m_transport.sendData(message.toBytes());
        } catch (IOException ioe) {
            throw new TransportException("IOException: "+ ioe.getMessage());
        }
    }

    /**
     * -------------------------------------------------------------------
     * Currently the same as sendData().
     */
    public void sendData(byte[] data, String dataType) throws TransportException {
        sendData(data);
    }

    /**
     * -------------------------------------------------------------------
     * Update the link id for this broker and the destination broker.
     */
    public void setLinkId(String stringId) {
        String oldID = m_linkID;
        m_linkID = stringId;

        m_parent.renameLink(oldID,m_linkID);

        if (oldID != null && oldID.length() > 0) {
            // Send an update link id request.
            LinkMessageCommand cmd =
		new LinkMessageCommand(LinkMessageCommand.UPDATE_CHANNEL_ID,m_linkID);
            cmd.setMessage(oldID);
            try {
                m_transport.sendData(cmd.toBytes());
            } catch (IOException ioe) {
                log.warn("IO error occurred while update link "+
                         oldID+" to new id "+stringId, ioe);
            }
        }
    }

    /**
     * -------------------------------------------------------------------
     */
    public Object getNaradaBrokeringId() {
        return m_brokerID;
    }

    /**
     * -------------------------------------------------------------------
     */
    public void setNaradaBrokeringId(Object brokeringId) {
        m_brokerID = brokeringId;
    }

    /**
     * -------------------------------------------------------------------
     */
    public String getLinkType() {
        return m_linkType;
    }

    /**
     * -------------------------------------------------------------------
     * Set a keep alive value for this link.  Links are killed off
     * by the link manager when keep alive has expired.
     */
    public void setKeepAliveTime(long keepAliveTime) {
        m_keepAliveStartTime = System.currentTimeMillis();
        m_keepAliveEndTime   = m_keepAliveStartTime + keepAliveTime;
        m_parent.updateKeepAliveLink(this);
    }

    /**
     * -------------------------------------------------------------------
     */
    public long getKeepAliveEndTime() {
        return m_keepAliveEndTime;
    }

    /**
     * -------------------------------------------------------------------
     */
    public String getLinkId() {
        return m_linkID;
    }

    /**
     * -------------------------------------------------------------------
     */
    public void closeLink() {
        m_closed = true;
        m_parent.closeLink(this);
    }

    /**
     * -------------------------------------------------------------------
     */
    public boolean isSecure() {
        return m_transport.isSecure();
    }

    /**
     * -------------------------------------------------------------------
     * @return The type of secure transport. [ SSL, ... ]
     */
    public String getSecurityInformation() {
        return m_linkType;
    }

    /**
     * -------------------------------------------------------------------
     */
    public void setLinkMigrationConstraint(String constraint, String linkType) {
        // TO DO 2
        // Link migration language.
    }

    /**
     * -------------------------------------------------------------------
     */
    public void setLinkStatusInterval(long interval) {
        m_transport.setStatusCheckInterval(interval);
    }

    /**
     * -------------------------------------------------------------------
     */
//    public void setPerformanceGathering(boolean gather) {
//        m_performanceEnabled = gather;
//        if (m_performanceEnabled && !gather) {
//            m_parent.getPerformanceMeasurement().removeMeasurementListener(this);
//        } else if (!m_performanceEnabled && gather) {
//            m_parent.getPerformanceMeasurement().addMeasurementListener(this);
//        }
//    }

    /**
     * -------------------------------------------------------------------
     */
//    public boolean performanceGatheringEnabled() {
//        return m_performanceEnabled;
//    }

    /**
     * -------------------------------------------------------------------
     */
//    public String[] getPerformanceFactorsMeasured() {
//        return PERF_MEASUREMENTS;
//    }

    /**
     * -------------------------------------------------------------------
     * @return The value of linkPerformanceData is "NA" if data not yet
     *      available.
     */
//    public LinkPerformanceData[] getAllPerformanceData() {
//        PerformanceMeasurement m = m_parent.getPerformanceMeasurement();
//
//        long currentTime = System.currentTimeMillis();
//        long startPeriod = currentTime - m.getMaxStatusInterval();
//
//        for(int i=0;i < PERF_STAT_TYPE.length;i++) {
//            String val = "NA";
//            try {
//                double v = m.getStat(getName(),
//                                    PERF_MEASUREMENT_NAMES[i],
//                                    startPeriod,
//                                    currentTime,
//                                    PERF_STAT_TYPE[i]);
//
//                if (Double.isInfinite(v)) {
//                    val = "Infinity";
//                } else {
//                    val = s_numberFormat.format(v);
//                }
//            } catch (IllegalArgumentException e) {
//            }
//            m_perfData[i].setParameterValue(val);
//            ((LinkPerformanceDataImpl)m_perfData[i]).setInterval((int)(m.getMaxStatusInterval()/1000));
//        }
//
//        return m_perfData;
//    }
//
    /**
     * -------------------------------------------------------------------
     * @return "NA" if data not yet available.  "NAN" if factor name is
     *      not understand.  Otherwise, value.
     */
//    public String getPerformanceData(String factorName) {
//        PerformanceMeasurement m = m_parent.getPerformanceMeasurement();
//        long currentTime = System.currentTimeMillis();
//        long startPeriod = currentTime - m.getMaxStatusInterval();
//
//        for(int i=0;i < PERF_MEASUREMENTS.length;i++) {
//            if (PERF_MEASUREMENTS[i].equals(factorName)) {
//                String val = "NA";
//                try {
//                    double v = m.getStat(getName(),
//                                        PERF_MEASUREMENT_NAMES[i],
//                                        startPeriod,
//                                        currentTime,
//                                        PERF_STAT_TYPE[i]);
//                    if (Double.isInfinite(v)) {
//                        val = "Infinity";
//                    } else {
//                        val = s_numberFormat.format(v);
//                    }
//                } catch (IllegalArgumentException e) {
//                }
//                return val;
//            }
//        }
//        return "NAN";
//    }

    /**
     * -------------------------------------------------------------------
     */
    public Object getBrokerID() {
        return m_brokerID;
    }

    /**
     * -------------------------------------------------------------------
     */
    public Transport getTransport() {
        return m_transport;
    }

    /**
     * -------------------------------------------------------------------
     * Sets the original properties that were passed into LinkFactory.createLink()
     * for this link.  Used primarily for link migration purposes.
     */
    public void setCreationProperties(Properties cp) {
        m_creationProperties = cp;
    }

    /**
     * -------------------------------------------------------------------
     * Gets the original properties that were passed into LinkFactory.createLink()
     * for this link.  Used primarily for link migration purposes.
     */
    public Properties getCreationProperties() {
        return new Properties(m_creationProperties);
    }

    /**
     * -------------------------------------------------------------------
     */
    public void handlePerformanceMessage(LinkMessage message) {
        byte[] payload = message.getPayload();

        String id = m_linkID;
        if (m_brokerID != null) {
            id = m_brokerID.toString();
        }
        byte[] measurementPayload = new byte[payload.length-1];
        System.arraycopy(payload,1,measurementPayload,0,payload.length-1);
        PerformanceMeasurement m = m_parent.getPerformanceMeasurement();
        switch(payload[0]) {
            case LinkMessagePerf.PERF_REQUEST:
                Vector result = m.getResponse(id,measurementPayload);
                Enumeration e = result.elements();
                while(e.hasMoreElements()) {
                    ByteArrayOutputStream baos = (ByteArrayOutputStream)e.nextElement();
                    payload = baos.toByteArray();
                    LinkMessagePerf newMessage = new LinkMessagePerf(m_linkID,LinkMessagePerf.PERF_RESPONSE);
                    newMessage.setPayload(payload);
                    try {
                        payload = newMessage.toBytes();
                        m_transport.sendData(payload);
                    } catch (IOException ioe) {
                        closeLink();
                    }
                }
                break;
            case LinkMessagePerf.PERF_RESPONSE:
                log.debug("Handling Performance Message["+payload[0]+"]");
                m.interpretResults(id,measurementPayload);
                break;
        }
    }

    /**
     * --------------------------------------------------------------------
     */
    public void performanceMessagesAvailable(Vector data) {
        //Log.debug("LinkImpl","Sending performance request: "+data.size());
        Enumeration e = data.elements();
        int i = 0;
        while(e.hasMoreElements()) {
            ByteArrayOutputStream baos = (ByteArrayOutputStream)e.nextElement();
            byte[] bytes = baos.toByteArray();

            // HACK ALERT
            // Update the send time to avoid the 20 second artificial delay
            long currentTime = System.currentTimeMillis();
            byte[] newTimeBytes = BytesUtil.longToBytes(currentTime);
            // Bytes 2 - 10 are the bytes for sent ping time.
            System.arraycopy(newTimeBytes,0,bytes,2,newTimeBytes.length);

            LinkMessagePerf msg = new LinkMessagePerf(m_linkID);
            msg.setPayload(bytes);
            i++;

            try {
                bytes = msg.toBytes();
                m_transport.sendData(bytes);
                Thread.currentThread().sleep(20);
            } catch (Exception ex) {
                closeLink();
                log.warn("IO error while sending performance messages",ex);
            }
        }
    }

    /**
     * --------------------------------------------------------------------
     */
    public long getStatusInterval() {
        return m_transport.getStatusCheckInterval();
    }

    /**
     * --------------------------------------------------------------------
     */
    public String getName() {
        if (m_brokerID == null) return m_linkID;
        return m_brokerID.toString();
    }

    /**
     * -------------------------------------------------------------------
     * This does not cause the link id change to be propagated to remote
     * brokers.  Only changes locally.
     */
    protected void updateLinkID(String id) {
        m_linkID = id;
    }
}
